package com.cx.base.day25_Rocketmq.producer.syncInfo;

import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

public class SyncBatchLargeFileProducer {
    public static void main(String[] args) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
        //        1.创建消息生产者，并且制定生产者组名
        DefaultMQProducer producer = new DefaultMQProducer("group1");
//        设置超时时间
        producer.setSendMsgTimeout(10000);
//        2.指定namesrv地址
        producer.setNamesrvAddr("192.168.48.128:9876;192.168.48.137:9876");
//        producer.setNamesrvAddr("47.96.165.93:9876;106.52.131.197:9876");
//        3.启动producer
        try {
            producer.start();
        } catch (
                MQClientException e) {
            e.printStackTrace();
        }
//      4.将消息对象放到集合中
        List<Message> msgs = new ArrayList<>();
        Message msg1 = new Message("batchTopic", "batch", ("helloworld" + 1).getBytes());
        Message msg2 = new Message("batchTopic", "batch1", ("helloworld" + 2).getBytes());
        Message msg3 = new Message("batchTopic", "batch2", ("helloworld" + 3).getBytes());
        msgs.add(msg1);
        msgs.add(msg2);
        msgs.add(msg3);
        BatchProducerLargeFileUtil batchConsumerLargeFileUtil = new BatchProducerLargeFileUtil(msgs);
        while (batchConsumerLargeFileUtil.hasNext()) {
            try {
                List<Message> listItem = batchConsumerLargeFileUtil.next();
                //      5.发送集合msgs
                SendResult result = producer.send(listItem);
                SendStatus status = result.getSendStatus();
                System.out.println("发送结果:" + status);
                TimeUnit.SECONDS.sleep(1);
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                //         关闭producer
                producer.shutdown();
            }
        }
    }
}

class BatchProducerLargeFileUtil implements Iterator<List<Message>> {
    private final int SIZE_LIMTI = 1024 * 1024 * 4;
    private final List<Message> messages;
    private int currIndex;

    public BatchProducerLargeFileUtil(List<Message> messages) {
        this.messages = messages;
    }

    public boolean hasNext() {
        return currIndex < messages.size();
    }

    public List<Message> next() {
        int nextIndex = currIndex;
        int totalSize = 0;
        for (; nextIndex < messages.size(); nextIndex++) {
            Message message = messages.get(nextIndex);
            //每条消息的数据大小=topic.length + message.getBody().length + message的属性值，即properties对象的存储数据量 + 日志20个字节
            int tmpSize = message.getTopic().length() + message.getBody().length; //topic.length + message.getBody().length
            Map<String, String> properties = message.getProperties();            //message的属性值，即properties对象的存储数据量
            for (Map.Entry<String, String> entry : properties.entrySet()) {
                tmpSize += entry.getKey().getBytes().length + entry.getValue().getBytes().length;
            }
            tmpSize += 20;  //日志20个字节                                       //日志20个字节
            if (tmpSize > SIZE_LIMTI) {
//                单个消息超过了最大的限度
//                忽略，否则会阻塞分裂进程
                if (nextIndex - currIndex == 0) {
//                    假如下一个子列表没有元素，则添加这个子列表然后退出循环，否则只是退出循环
                    nextIndex++;
                }
                break;
            }
            if (tmpSize + totalSize > SIZE_LIMTI) {
                break;
            } else {
                totalSize += tmpSize;
            }
        }

        List<Message> subList = messages.subList(currIndex, nextIndex);
        currIndex = nextIndex;
        return subList;
    }
}
